package flink_p3_cep

import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.windowing.time.Time



case class LoginEvent(id:Int, name:String, status:String, time:Long)

object cep01_loginTest {


  //检测恶意登录：10秒钟内失败3次即为恶意登录
  def main(args: Array[String]): Unit = {

    import org.apache.flink.streaming.api.scala._


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val dataStream: DataStream[LoginEvent] = env.fromCollection(List(
      new LoginEvent(1, "zs", "F", 1000),
      new LoginEvent(2, "ls", "F", 2000),
      new LoginEvent(3, "zs", "F", 3000),
      new LoginEvent(4, "ls", "F", 4000),
      new LoginEvent(5, "zs", "S", 5000),
      new LoginEvent(6, "ls", "F", 6000),
      new LoginEvent(7, "zs", "S", 7000),
      new LoginEvent(8, "ls", "F", 8000)
    )).assignAscendingTimestamps(_.time)


    val pattern: Pattern[LoginEvent, LoginEvent] = Pattern.begin[LoginEvent]("start")
      .where(_.status.equals("F"))          //登录失败的
      .timesOrMore(3)                 //  出现3次或则以上
      .greedy                               // 贪婪模式  尽可能多的匹配
      .within(Time.seconds(10))    // 贪婪模式也有个时间夏限度


    val ps: PatternStream[LoginEvent] = CEP.pattern(dataStream.keyBy(_.name), pattern)

    val resStream: DataStream[String] = ps.select(
      patternSelectFunction => {
        val list: List[LoginEvent] = patternSelectFunction.get("start").get.toList
        val sb = new StringBuilder
        for (i <- list.indices) {
          sb.append(s"${list(i).name} 第 $i 次登录失败, time: ${list(i).time}  ")
        }
        sb.toString()
      }
    )


    resStream.print()


    env.execute("cep demo 1")





  }



}
